package cn.chuengxu._02_asyn;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**======================================================================================
* @Title:
* @Description: 异步发送消息消费者
* @Author: Mr.Zhang
* @Version: 1.0
* @Date:  2022/7/25 12:21
* @Parameters:
* @Return
======================================================================================*/
public class Consumer {

    public static void main(String[] args) throws Exception {
        // 1.创建消费者对象
        DefaultMQPushConsumer syncConsumer = new DefaultMQPushConsumer("sync_Consumer_hello");

        // 2.连接mq
        syncConsumer.setNamesrvAddr("127.0.0.1:9876");

        // 3.订阅消息
        syncConsumer.subscribe("asyn_Hello","*");

        // 4.注册监听器
        syncConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (Message message:msgs){
                    String result = new String(message.getBody(), StandardCharsets.UTF_8);
                    System.out.println(result);
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 5.启动消费者
        syncConsumer.start();
    }


}
